package org.wejob.core.log;

import io.netty.util.internal.shaded.org.jctools.queues.SpscLinkedQueue;
import org.wejob.core.log.util.BufferPool;
import org.wejob.core.log.util.SystemTime;
import org.wejob.core.log.util.mpscqueue.MpscArrayQueue;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

public class Topic {
    int recordNUms;
    int position;
    String name;
    MappedByteBuffer queue;
    RandomAccessFile messageFile;
    FileChannel fileChannel;
    //异步提交任务队列
    MpscArrayQueue<MessageRecord> messageQueue;
    Mqueue mq;
    ByteBuffer memoryPage;
    MemoryRecords curbatch;
    volatile long maxCommitedOffset;
    long maxCodedOffset;
    AtomicLong nextOffset;
    AtomicBoolean codeLock;
    AtomicBoolean fileLock;
    SpscLinkedQueue<MemoryRecords> toWriteTasks;
    BufferPool bufferPool;
    long logsize;
    IndexOffset indexOffset;


    public Topic(String name) throws IOException {
        this(name,0);
    }
    public Topic(String name,int logsize) throws IOException {
        this.codeLock=new AtomicBoolean(false);
        this.fileLock=new AtomicBoolean(false);
        this.nextOffset=new AtomicLong(0);
        this.name=name;
        this.messageFile= new RandomAccessFile(this.name,"rw");
        this.messageFile.setLength(1024*1024*1024);
        this.fileChannel=messageFile.getChannel();
        this.queue=fileChannel.map(FileChannel.MapMode.READ_WRITE,0,this.messageFile.length());
        this.messageQueue=new MpscArrayQueue<>(1024*1024);
        this.toWriteTasks=new SpscLinkedQueue<>();
        this.bufferPool=new BufferPool(32*1024*1024,4*1024,new SystemTime(),"bufferpool");
        this.logsize=logsize;
        this.indexOffset=new IndexOffset(this);
    }


    public void close() throws IOException {
        this.queue.force();
        this.fileChannel.force(true);
        this.messageFile.close();
    }

    public long offsetCounter(){
        return this.nextOffset.getAndIncrement();
    }
    public int writeAndCode() throws IOException, InterruptedException {
        doWrite();
        int var2=doCode();
        if(var2==-1){
            doWrite();
        }
        return var2;
    }
    public int doCode() throws IOException, InterruptedException {
        //每次贪心处理topic中的消息,直到写满一个内存页或者队列为空
        if(this.tryCodeLock()){
            for(;;){
                MessageRecord record=this.messageQueue.poll();
                if(record==null){
                    releaseCodeLock();
                    return 0;
                }
                if(this.curbatch ==null){
                    ByteBuffer byteBuffer =bufferPool.allocate(BufferPool.pagesize,1000);
                    this.curbatch=new MemoryRecords(byteBuffer,0);
                }
                byte[] serializeMessage=serializeMessage(record);
                int size=serializeMessage.length;
                //当前缓存页足够时直接写入
                if(serializeMessage.length<=this.curbatch.buffer.limit()-this.curbatch.buffer.position()){
                    this.curbatch.append(record.offset,serializeMessage);
                    if(record.callback!=null){
                        if(this.curbatch.callbacks==null){
                            this.curbatch.callbacks=new LinkedList<>();
                        }
                        this.curbatch.callbacks.add(record.callback);
                    }
                    this.curbatch.records.add(record);
                    //this.maxCodedOffset= record.offset;
                }else if(size> this.bufferPool.poolableSize()){//size 大于pagesize
                    //MemoryRecords batch
                    this.toWriteTasks.offer(this.curbatch);
                    this.curbatch=null;
                    System.out.println("to write batch1--"+this.curbatch);
                    ByteBuffer byteBuffer=this.bufferPool.allocate(size,1000);
                    byteBuffer.put(serializeMessage);
                    MemoryRecords batch2=new MemoryRecords(byteBuffer,record.offset);
                    batch2.records.add(record);
                    this.toWriteTasks.offer(batch2);
                    System.out.println("to write batch2--"+this.curbatch);
                    if(record.callback!=null){
                        batch2.callbacks=new LinkedList<>();
                        batch2.callbacks.add(record.callback);
                    }
                    releaseCodeLock();
                    return -1;
                }else{
                    this.toWriteTasks.offer(curbatch);
                    ByteBuffer byteBuffer=this.bufferPool.allocate(this.bufferPool.poolableSize(),1000);
                    this.curbatch=new MemoryRecords(byteBuffer,0);
                    this.curbatch.append(record.offset,serializeMessage);
                    this.curbatch.records.add(record);

                    if(record.callback!=null){
                        if(this.curbatch.callbacks==null){
                            this.curbatch.callbacks=new LinkedList<>();
                        }
                        this.curbatch.callbacks.add(record.callback);
                    }
                    releaseCodeLock();
                    return -1;
                }

            }
        }
        //todo 这里应该返回0还是-1;
        return 0;
    }

    public void doWrite() throws IOException {
        if(tryFileLock()){
            for(;;){

                MemoryRecords var1=this.toWriteTasks.poll();
                if(var1==null) {
                    releaseFileLock();
                    return;
                }
                int wiritten=var1.writeTo(this.fileChannel);
                this.indexOffset.append(var1.maxOffset, this.logsize);
                this.logsize+=wiritten;
                this.bufferPool.deallocate(var1.buffer);
                //todo 执行回调逻辑--交给线程池
                if(var1.callbacks!=null){
                    var1.callbacks.forEach(callback -> {
                        new Thread(()->{
                            callback.onCompletion("success",null);
                        }).start();
                    });
                }
                this.maxCommitedOffset=var1.maxOffset;//防止指令重排,赋值操作必须在获取锁之前完成
                var1.records.forEach(messageRecord -> {
                    if(messageRecord.future!=null){
                        messageRecord.future.futureLock.lock();
                        messageRecord.future.result="success";
                        if(messageRecord.future.waiter!=null){
                            messageRecord.future.waiter.signal();
                        }

                        messageRecord.future.futureLock.unlock();
                    }

                });


            }

        }
    }
    public boolean tryCodeLock(){
        return this.codeLock.compareAndSet(false,true);
    }
    public void releaseCodeLock(){
        this.codeLock.set(false);
    }
    public boolean tryFileLock(){
        return this.fileLock.compareAndSet(false,true);
    }
    public void releaseFileLock(){
        this.fileLock.set(false);
    }
    private byte[] serializeMessage(MessageRecord message){
        //message.
        return (message.offset+message.message().toString()).getBytes(StandardCharsets.UTF_8);
    }

}
